package mapreduce

import "container/list"
import (
	"fmt"
)

const (
	STATUS_IDLE    int = 1
	STATUS_WORKING int = 2
)

type WorkerInfo struct {
	address string
	status  int
	// You can add definitions here.
}

// Clean up all workers by sending a Shutdown RPC to each one of them Collect
// the number of jobs each work has performed.
func (mr *MapReduce) KillWorkers() *list.List {
	l := list.New()
	for _, w := range mr.Workers {
		DPrintf("DoWork: shutdown %s\n", w.address)
		args := &ShutdownArgs{}
		var reply ShutdownReply
		ok := call(w.address, "Worker.Shutdown", args, &reply)
		if ok == false {
			fmt.Printf("DoWork: RPC %s shutdown error\n", w.address)
		} else {
			l.PushBack(reply.Njobs)
		}
	}
	return l
}

func (mr *MapReduce) RunMaster() *list.List {
	// Your code here
	// 1. wait for worker start up already
	go func() {
		for mr.alive {
			worker := <-mr.registerChannel
			mr.Workers[worker] = &WorkerInfo{address: worker, status: STATUS_IDLE}
		}
	}()
	for mr.alive {
		if worker := getIdleWorker(mr); worker == nil {
			continue
		} else {
			if mr.nUsedMap != mr.nMap {
				worker.status = STATUS_WORKING
				args := &DoJobArgs{}
				args.Operation = Map
				args.JobNumber = mr.nUsedMap
				args.File = mr.file
				args.NumOtherPhase = mr.nReduce

				var reply DoJobReply
				ok := call(worker.address, "Worker.DoJob", args, &reply)
				if ok == false {
					fmt.Printf("Register: RPC %s DoJob error\n", worker.address)
				} else {
					mr.nUsedMap = mr.nUsedMap + 1
				}
				worker.status = STATUS_IDLE
				continue
			}

			if mr.nUserReduce == mr.nReduce {
				break
			} else {
				worker.status = STATUS_WORKING
				args := &DoJobArgs{}
				args.Operation = Reduce
				args.JobNumber = mr.nUserReduce
				args.File = mr.file
				args.NumOtherPhase = mr.nMap

				var reply DoJobReply
				ok := call(worker.address, "Worker.DoJob", args, &reply)
				if ok == false {
					fmt.Printf("Register: RPC %s DoJob error\n", worker.address)
				} else {
					mr.nUserReduce = mr.nUserReduce + 1
				}
				worker.status = STATUS_IDLE

			}
		}
	}
	// 2. send worker map job and wait for all already
	// 3. send worker reduce job and wait for all already
	return mr.KillWorkers()
}

func getIdleWorker(mr *MapReduce) *WorkerInfo {
	for _, w := range mr.Workers {
		if w.status == STATUS_IDLE {
			return w
		}
	}
	return nil
}
